Conversation
b07d581 to
96eafdf
Compare
This is a thorough piece of work. Reusing A few things I would like resolved before merge. No performance data for the sort path
The docs change says the Tracking issue for the nested case Same question for the nested array and struct divergence the docs now call out. There is no issue link in the new Sort-merge join builds an expression it throws away At Silent passthrough in
One test question In |
|
Updated in 38825e26a. The follow-up adds the sort benchmark and links the two remaining compatibility issues from the guide. I checked the six points separately:
The new release benchmark built, all eight smoke cases passed, and benchmark Clippy ( |
|
One thing to consider is https://issues.apache.org/jira/browse/SPARK-54918 which fixes +0, -0 in Spark 4.2.0 |
|
@comphead Thanks for the pointer. I checked the SPARK-54918 patch. It extends normalization to array set operations such as For the scalar sorting and ranking cases here, Spark already treats This PR normalizes the comparison keys consistently across sorting and ranking, so the fix is still needed with Spark 4.2. |
38825e2 to
552fc51
Compare
`CometSortOrder.getSupportLevel` delegated to the recursive `SupportLevel.strictFloatingPointReason`, which rejects a data type that contains a float or double at any nesting level. Under `spark.comet.exec.strictFloatingPoint=true` that rejected scalar FLOAT and DOUBLE sort keys as well as nested ones. Scalar floating-point comparison keys are normalized before native Sort, TopK, Window, WindowGroupLimit, and range partitioning (apache#5469): NaN payloads are folded together and signed zeros tied, matching Spark's SQLOrderingUtil.compareDoubles/compareFloats. Only the comparison key is normalized, so returned values keep their original NaN representation and zero sign. Those keys are therefore compatible even in strict mode. Match scalar FLOAT/DOUBLE ahead of the shared helper so they report Compatible, and leave every other data type on the existing recursive path. Floats nested in arrays, structs, and maps still sort by Arrow's raw total ordering, under which -0.0 sorts below 0.0 and a sign-bit NaN sorts below -Infinity, so they keep falling back (apache#5507). The three other callers of `strictFloatingPointReason` are untouched, including `SortArray`, so `sort_array` on a floating-point element type still falls back. The description of `spark.comet.exec.strictFloatingPoint` is updated to say which sorts are still affected. It is scraped into the generated config reference, so it is the only warning many users will read. The subject of the fallback reason is shared with `getIncompatibleReasons()` through one private val, since that text is published per Spark version in the generated compatibility pages and the two must not drift. Tests: the existing scalar test now asserts native execution instead of fallback, and sorts on a unique id last. -0.0 and +0.0 are peers under Spark's comparison, so ordering on the float columns alone leaves ties whose relative order neither engine promises, and Comet and Spark already lay those out differently today with strict mode off. The sibling array and struct tests still assert the strict-mode fallback. A generated matrix covers both types, both directions, both null orderings and compound keys, one test pins that the sort leaves NaN payloads and zero signs untouched, and one covers the TopK path, which reaches the same gate but a different operator. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The floating-point compatibility page still carried apache#5469's note that the strict policy was unchanged for scalar sort keys, pointing at apache#5506. The tuning page stated flatly that sorting on floating-point data is not compatible with Spark, which was already wrong for scalar keys after apache#5469. Both now scope the claim to ordering keys. `ORDER BY`, window ordering and range partitioning on scalar FLOAT and DOUBLE stay native under strict mode. Values nested in arrays, structs and maps still fall back, and so does `sort_array` on a floating-point element type, which sorts elements rather than ordering rows and is gated separately. Neither page is generated. `GenerateDocs` emits the per-expression compatibility pages and the config reference, so those pick up the source changes in this branch on their own. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
`CometSortOrder.getSupportLevel` delegated to the recursive `SupportLevel.strictFloatingPointReason`, which rejects a data type that contains a float or double at any nesting level. Under `spark.comet.exec.strictFloatingPoint=true` that rejected scalar FLOAT and DOUBLE sort keys as well as nested ones. Scalar floating-point comparison keys are normalized before native Sort, TopK, Window, WindowGroupLimit, and range partitioning (apache#5469): NaN payloads are folded together and signed zeros tied, matching Spark's SQLOrderingUtil.compareDoubles/compareFloats. Only the comparison key is normalized, so returned values keep their original NaN representation and zero sign. Those keys are therefore compatible even in strict mode. Match scalar FLOAT/DOUBLE ahead of the shared helper so they report Compatible, and leave every other data type on the existing recursive path. Floats nested in arrays, structs, and maps still sort by Arrow's raw total ordering, under which -0.0 sorts below 0.0 and a sign-bit NaN sorts below -Infinity, so they keep falling back (apache#5507). The three other callers of `strictFloatingPointReason` are untouched, including `SortArray`, so `sort_array` on a floating-point element type still falls back. The description of `spark.comet.exec.strictFloatingPoint` is updated to say which sorts are still affected. It is scraped into the generated config reference, so it is the only warning many users will read. The subject of the fallback reason is shared with `getIncompatibleReasons()` through one private val, since that text is published per Spark version in the generated compatibility pages and the two must not drift. Tests: the existing scalar test now asserts native execution instead of fallback, and sorts on a unique id last. -0.0 and +0.0 are peers under Spark's comparison, so ordering on the float columns alone leaves ties whose relative order neither engine promises, and Comet and Spark already lay those out differently today with strict mode off. The sibling array and struct tests still assert the strict-mode fallback. A generated matrix covers both types, both directions, both null orderings and compound keys, one test pins that the sort leaves NaN payloads and zero signs untouched, and one covers the TopK path, which reaches the same gate but a different operator.
The floating-point compatibility page still carried apache#5469's note that the strict policy was unchanged for scalar sort keys, pointing at apache#5506. The tuning page stated flatly that sorting on floating-point data is not compatible with Spark, which was already wrong for scalar keys after apache#5469. Both now scope the claim to ordering keys. `ORDER BY`, window ordering and range partitioning on scalar FLOAT and DOUBLE stay native under strict mode. Values nested in arrays, structs and maps still fall back, and so does `sort_array` on a floating-point element type, which sorts elements rather than ordering rows and is gated separately. Neither page is generated. `GenerateDocs` emits the per-expression compatibility pages and the config reference, so those pick up the source changes in this branch on their own.
…ache#5981) * feat: admit scalar floating-point sort keys under strict floating point `CometSortOrder.getSupportLevel` delegated to the recursive `SupportLevel.strictFloatingPointReason`, which rejects a data type that contains a float or double at any nesting level. Under `spark.comet.exec.strictFloatingPoint=true` that rejected scalar FLOAT and DOUBLE sort keys as well as nested ones. Scalar floating-point comparison keys are normalized before native Sort, TopK, Window, WindowGroupLimit, and range partitioning (apache#5469): NaN payloads are folded together and signed zeros tied, matching Spark's SQLOrderingUtil.compareDoubles/compareFloats. Only the comparison key is normalized, so returned values keep their original NaN representation and zero sign. Those keys are therefore compatible even in strict mode. Match scalar FLOAT/DOUBLE ahead of the shared helper so they report Compatible, and leave every other data type on the existing recursive path. Floats nested in arrays, structs, and maps still sort by Arrow's raw total ordering, under which -0.0 sorts below 0.0 and a sign-bit NaN sorts below -Infinity, so they keep falling back (apache#5507). The three other callers of `strictFloatingPointReason` are untouched, including `SortArray`, so `sort_array` on a floating-point element type still falls back. The description of `spark.comet.exec.strictFloatingPoint` is updated to say which sorts are still affected. It is scraped into the generated config reference, so it is the only warning many users will read. The subject of the fallback reason is shared with `getIncompatibleReasons()` through one private val, since that text is published per Spark version in the generated compatibility pages and the two must not drift. Tests: the existing scalar test now asserts native execution instead of fallback, and sorts on a unique id last. -0.0 and +0.0 are peers under Spark's comparison, so ordering on the float columns alone leaves ties whose relative order neither engine promises, and Comet and Spark already lay those out differently today with strict mode off. The sibling array and struct tests still assert the strict-mode fallback. A generated matrix covers both types, both directions, both null orderings and compound keys, one test pins that the sort leaves NaN payloads and zero signs untouched, and one covers the TopK path, which reaches the same gate but a different operator. * feat: admit scalar floating-point range partitioning under strict mode `supportedRangePartitioningDataType` rejected FLOAT and DOUBLE whenever `spark.comet.exec.strictFloatingPoint=true`. This is a second admission gate, independent of `CometSortOrder`, so narrowing only the sort gate left a plan that sorts natively but still shuffles on the JVM: `CometSort` over `CometColumnarExchange` rather than `CometExchange`/`CometNativeShuffle`. The native range partitioner normalizes both sides of its comparison. Incoming keys are the serialized sort-order expressions, which carry the `NormalizeNaNAndZero` wrapper, and the sampled boundary rows are normalized before being row-encoded with the same converter. Scalar floating-point keys therefore partition consistently with Spark's ordering, so the strict-mode rejection buys no correctness. Accept scalar FLOAT and DOUBLE, and drop the now-unreachable strict-mode fallback reason. Nested types are unaffected: they already fall through to the catch-all that rejects them. The two policy tests are merged into one parameterized over strict mode, since the expected outcome no longer depends on it. The merged test also drops `SortOrder.allowIncompatible=true`, which the old strict-mode test needed to reach this gate, so it now proves the whole path works without an escape hatch. * test: cover the other paths the sort-order gate admits `CometSortOrder` is the admission gate for nine serde sites, so narrowing it for scalar floating point opens more than `ORDER BY`. Each of these was verified rather than assumed. Sort-merge join synthesizes sort orders and therefore reaches the same gate, which means strict mode used to make FP-keyed joins fall back as a side effect. Comet does not normalize the join keys itself. It compares `join_on`, and correctness rests on Catalyst's NormalizeFloatingNumbers having already wrapped both sides, so the new test pins that dependency. Its fixtures are local relations rather than Parquet tables because a Parquet round trip canonicalizes every NaN payload, which would collapse the two NaN cases into one. The WindowGroupLimit peer test is parameterized over strict mode, and a second window test is added that is deliberately not gated on Spark 3.5, so window ordering has coverage on every supported version. It orders by a floating point column and sums over the default RANGE frame, which spans the peer group, so -0.0 and +0.0 being peers is visible in the result and the expected values do not depend on tie order. Columnar shuffle performs range partitioning on the JVM with Spark's own RangePartitioner, but still probes whether Comet can serialize the sort order, so strict mode used to abandon an exchange it would have executed identically. The new test pins the floating-point half of that. Consulting a native-serde gate on a path that never goes native is tracked in apache#5971. * docs: describe which floating-point sorts strict mode still affects The floating-point compatibility page still carried apache#5469's note that the strict policy was unchanged for scalar sort keys, pointing at apache#5506. The tuning page stated flatly that sorting on floating-point data is not compatible with Spark, which was already wrong for scalar keys after apache#5469. Both now scope the claim to ordering keys. `ORDER BY`, window ordering and range partitioning on scalar FLOAT and DOUBLE stay native under strict mode. Values nested in arrays, structs and maps still fall back, and so does `sort_array` on a floating-point element type, which sorts elements rather than ordering rows and is gated separately. Neither page is generated. `GenerateDocs` emits the per-expression compatibility pages and the config reference, so those pick up the source changes in this branch on their own. * test: drop redundant string interpolator in range partitioning test name
3 participants
Why are the changes needed?
A filter such as
RANK() <= 1must keep every row tied for first place. Native execution can currently drop some of those rows when the ordering column contains signed zeros or different NaN representations. This PR fixes that result mismatch for scalarFLOATandDOUBLEkeys, following the nativeWindowGroupLimitExecsupport added in #4870.A rank cutoff can discard a tied row
Consider a table
measurementswhoseDOUBLEcolumnvcontains-0.0,+0.0, and1.0:Spark considers the two zeros equal, so both belong at rank 1. The native comparison instead distinguishes their IEEE-754 representations:
-0.0+0.01.0The query therefore keeps both zero rows in Spark, but only the negative-zero row before this fix.
DENSE_RANKhas the same problem at the cutoff. OnceWindowGroupLimitExechas discarded a qualifying row, the window calculation above it cannot recover that row.NaNs expose the same mismatch. Spark treats every NaN representation as equal and greater than every non-NaN value. For input containing two differently encoded NaNs and
1.0,ORDER BY v DESCfollowed byrnk <= 1should keep both NaNs. Arrow's raw total ordering can separate them; a NaN with its sign bit set can even sort below finite values. The regressions in this PR construct distinct NaN encodings directly so the test actually exercises this distinction.What changes were proposed in this PR?
The fix gives the native sorting and window operators a common representation for comparing scalar floating values. It reuses Comet's existing
NormalizeNaNAndZeroexpression to map every NaN to one canonical NaN and both zero signs to positive zero in the comparison keys. Arrow can then compare those keys using its existing machinery while agreeing with Spark about which values are tied.This normalization must happen before sorting as well as when assigning ranks. For example, with
ORDER BY v, secondary, the rows(-0.0, 1)and(+0.0, 1)are peers and must precede the rows whose secondary key is2. A sort that distinguishes the zero signs can instead produce:Even a corrected peer comparison would see the first peer group split apart. A streaming rank limit could reach the cutoff at
(-0.0, 2)and stop before seeing the other qualifying row. The shared comparison-key construction therefore supplies normalized keys to Sort, Window, and WindowGroupLimit, keeping their ordering and peer decisions consistent.That consistency includes
PARTITION BY, not justORDER BY. An ordinary query such asRANK() OVER (PARTITION BY p ORDER BY id)must still work whenpis aDOUBLEcolumn containing only1.0and2.0. Spark may already have normalizedp; wrapping it again in Sort while leaving Window unchanged produces different expressions for the same key. DataFusion uses those expressions to recognize partition ordering, so the mismatch causes execution to fail even though the values compare equally. The planner therefore reuses already-normalized expressions and constructs matching keys for Sort, Window, and WindowGroupLimit. This preserves ordinary windows as well as the rank-limit path.Native range partitioning uses the same rule for its sampled boundaries. An incoming positive zero and a boundary containing negative zero must compare equal, just as they do during sorting. Normalizing both sides keeps the shuffle's comparisons consistent with the operators that consume its output.
The original input values are preserved: selecting
vstill returns its original zero sign or NaN payload. Only the temporary keys used for comparison are normalized. The scope remains scalarFLOATandDOUBLE; the documented limitations for floats nested in arrays or structs and the existingspark.comet.exec.strictFloatingPointfallback policy are unchanged.How was this PR tested?
The correctness results in the following paragraphs and table were recorded during the earlier implementation and rebase. They are historical local validation, not a fresh full-suite rerun for this review follow-up. The newly run sort benchmark and checks are described after those results.
The regressions verify both the returned rows and the execution path. The Spark cutoff tests require native Sort and WindowGroupLimit operators, compare results with Spark, and check that the input and output retain their original floating bits. Native tests cover compound keys, peer groups spanning batches, and agreement between range boundaries and incoming keys. The previously ignored signed-zero SQL regression is also enabled.
Additional regressions protect ordinary windows partitioned by floating columns:
RANK,PERCENT_RANK, andNTILEmust match Spark for single and compound partitions. A filteredRANKquery also checks the Sort → WindowGroupLimit → Window path. A native planner test verifies that both bare and already-normalized partition keys retain recognizable ordering metadata.With the comparison-key fix removed, the native peer/range regressions and both Spark FLOAT/DOUBLE cutoff regressions fail. In the Spark fixture, unpatched Comet returns 4 rows where Spark returns 8. The fixed runs pass.
On Spark 4.1, the partition-ordering cases were also checked against the library before the key-consistency fix: all six ordinary-window cases failed, while the two WindowGroupLimit controls passed. All eight pass with the fix.
The final review reran all three native planner regressions and the eight Spark 4.1 partition-key regressions; all passed. It also checked the three query shapes behind the Spark SQL CI failures on the earlier revision: a count window partitioned by floating columns, a rank window with computed aggregate/partition keys, and a correlated
EXISTS ... LIMIT 1query that Spark rewrites into a window. Each query matched Spark with native Sort and Window execution asserted. The same probes passed with the pinned base library. These were focused query probes, not reruns of the full CI shards.make core,cargo fmt --all -- --check, andcargo clippy --all-targets --workspace -- -D warningspassed. JVM formatting checks and Spark 4.1 Scala style checks also passed.The review follow-up adds a reproducible scalar floating-key sort benchmark. It executes DataFusion 54.1.0
SortExecwith a bareColumnkey and with Comet'sNormalizeNaNAndZerokey in the same optimized binary, representing the before/after key-construction change for previously bare keys. Already-normalized keys are reused by the planner. This measures execution and output draining for an in-memory, single-partition sort; it does not measure Spark planning, JNI, I/O, or a distributed range shuffle. Input generation and physical-plan construction are outside the timed region.Both variants sort 262,144 rows in 32 batches of 8,192, with one current-thread Tokio runtime, ascending/nulls-last ordering, and no fetch limit. The finite shape has no nulls; the mixed shape has approximately 20% nulls, 10% payload/sign NaNs, 10% signed zeros, and 60% finite values. The release profile uses thin LTO and one codegen unit. Measurements used an AMD EPYC-Milan CPU pinned to one core, Rust 1.98.0, 30 Criterion samples per case, a one-second warmup, and a three-second measurement target. Other workspace builds and tests were paused. A second pass reversed the order of the two variants.
Each cell reports bare → normalized even when execution order was reversed. FLOAT differences were small and the mixed-shape direction changed between passes. DOUBLE normalization had a measurable cost in both passes: +6.6–11.4% for finite values and +5.2–12.5% for mixed values. These are observed repeat ranges, not confidence intervals or end-to-end Spark performance claims. Normalization deliberately changes the comparison semantics for the mixed input. Folding it into sorting or encoding could avoid temporary arrays, but that optimization is not implemented by this correctness fix.
Untimed validation checks every row identity, original floating bits and nulls, ordering, and zero spills. Peak DataFusion pool reservations were identical for bare and normalized keys: 16 MiB / 16.0625 MiB for FLOAT finite/mixed and 18 MiB / 18.0625 MiB for DOUBLE finite/mixed. These are reserved-pool peaks, not process RSS or a complete allocation high-water mark: they do not include the temporary normalized values buffers. Each normalization evaluation allocates 4 bytes per row for FLOAT or 8 for DOUBLE (32/64 KiB for an 8,192-row array), sharing the validity bitmap. That is a per-evaluation size, not total allocation traffic; sorting and merging may evaluate the key more than once.
To reproduce from
native/:For this follow-up, the release benchmark build, all eight benchmark smoke cases, release Clippy for this benchmark with
-D warnings,cargo fmt --all -- --check, andgit diff --checkpassed. The broader correctness suites above were not rerun for this benchmark/documentation-only commit.The guide now links #5506 for narrowing strict-mode admission of corrected scalar sort keys and #5507 for nested ORDER BY/rank compatibility. Spark 4.1.3 diagnostic queries reproduced nested array/struct ORDER BY and array RANK mismatches for both FLOAT and DOUBLE; struct RANK already fell back and matched Spark. All eight strict-mode controls matched Spark, and four scalar planning controls confirmed the conservative strict-mode fallback. Those tracker probes reused a prior native build with the unchanged nested path and matching JVM admission sources; they are not a full rebuild or correctness-suite run of this PR head.
Which issue does this PR close?
Closes #5468.